use std::time::Duration;

use proto::types::{
    ACK, NOTIFY, NOTIFY_ACK, PING, PONG, REQUEST, RESPONSE, SIMPLE_REQUEST, SIMPLE_RESPONSE,
};

use crate::{
    api::{on_notify, on_request, on_simple_request},
    AppErr,
};

use super::{client::Client, conn_map::ConnMap, frame::Frame, reader::ConnReader};

async fn read_wait_exit(client: &Client, reader: &mut ConnReader) -> Result<Frame, AppErr> {
    tokio::select! {
        frame = reader.read_with_timeout(Duration::from_secs(10)) => {
            frame
        }

        _ = client.wait_read_exit() => {
            Err(AppErr::Timeout("read timeout"))
        }
    }
}

async fn read_loop(client: &Client, reader: &mut ConnReader) -> Result<(), AppErr> {
    loop {
        let frame = read_wait_exit(&client, reader).await?;

        match frame.ts() {
            ACK | PONG | SIMPLE_RESPONSE | RESPONSE => {
                client.dispatch(frame);
            }

            NOTIFY_ACK => {
                client.min_frame(ACK, frame.seq()).await?;
                tokio::spawn(on_notify(frame));
            }

            NOTIFY => {
                tokio::spawn(on_notify(frame));
            }

            REQUEST => {
                client.min_frame(ACK, frame.seq()).await?;
                tokio::spawn(on_request(frame, client.clone()));
            }

            SIMPLE_REQUEST => {
                client.min_frame(ACK, frame.seq()).await?;
                tokio::spawn(on_simple_request(frame, client.clone()));
            }

            PING => {
                client.min_frame(PONG, frame.seq()).await?;
            }

            _ => {
                return Err(AppErr::Invalid("invalid ts"));
            }
        };
    }
}

async fn read_task(client: Client, mut reader: ConnReader, conn_map: ConnMap) {
    match read_loop(&client, &mut reader).await {
        Ok(()) => {}
        Err(e) => {
            println!("read exit:{}", e);
        }
    }
    client.close().await;
    conn_map.remove(client.addr());
}
